package net.bwie.douyin.dwd.log.Function;


import net.bwie.douyin.dwd.log.bean.LiveEvent;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class OnlineUsersFunction implements WindowFunction<LiveEvent, String, String, TimeWindow> {

    private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

    @Override
    public void apply(String s, TimeWindow window, Iterable<LiveEvent> input, Collector<String> out) throws Exception {
        String windowStartTime = format.format(window.getStart());
        String windowEndTime = format.format(window.getEnd());

        int uv_count = 0;

        for (LiveEvent liveEvent : input) {
            if ("enter".equals(liveEvent.getEventType())){
                uv_count++;
            }
            if ("leave".equals(liveEvent.getEventType())){
                uv_count--;
            }
        }


       out.collect(windowStartTime + "," + windowEndTime +  "," + uv_count+","+System.currentTimeMillis());
    }
}
